package com.kai.mqtt.message.service.impl;

import com.kai.mqtt.data.PageResult;
import com.kai.mqtt.data.request.PageSubscribeMessageRequest;
import com.kai.mqtt.data.response.SubscribeMessageResponse;
import com.kai.mqtt.message.ShareSubscribeMessage;
import com.kai.mqtt.message.SubscribeMessage;
import com.kai.mqtt.message.service.ISubscribeMessageService;
import com.kai.mqtt.util.TopicUtil;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Stream;

/**
 * @author weijk
 * @since 2024/6/10
 */
@Service
public class SubscribeMessageServiceImpl implements ISubscribeMessageService {

    private static final Map<String, Map<String, SubscribeMessage>> MAP = new ConcurrentHashMap<>();

    private static final Map<String, ShareSubscribeMessage> SHARE_MAP = new ConcurrentHashMap<>();

    @Override
    public void put(String topic, SubscribeMessage subscribeMessage) {
        if (TopicUtil.isShareTopic(topic)) {
            ShareSubscribeMessage shareSubscribeMessage = SHARE_MAP.computeIfAbsent(topic, (key) -> new ShareSubscribeMessage());
            shareSubscribeMessage.add(subscribeMessage);
        } else {
            Map<String, SubscribeMessage> map = MAP.computeIfAbsent(topic, (key) -> new ConcurrentHashMap<>());
            map.put(subscribeMessage.getClientId(), subscribeMessage);
        }
    }

    @Override
    public void remove(String clientId) {
        MAP.forEach((topic, messageMap) -> {
            messageMap.remove(clientId);
            if (messageMap.isEmpty()) {
                MAP.remove(topic);
            }
        });

        SHARE_MAP.forEach((topic, message) -> {
            message.remove(clientId);
            if (message.isEmpty()) {
                SHARE_MAP.remove(topic);
            }
        });
    }

    @Override
    public void remove(String topic, String clientId) {
        Map<String, SubscribeMessage> map = MAP.get(topic);
        if (Objects.nonNull(map)) {
            map.remove(clientId);
            if (map.isEmpty()) {
                MAP.remove(topic);
            }
        }

        ShareSubscribeMessage shareSubscribeMessage = SHARE_MAP.get(topic);
        if (Objects.nonNull(shareSubscribeMessage)) {
            shareSubscribeMessage.remove(clientId);
            if (shareSubscribeMessage.isEmpty()) {
                SHARE_MAP.remove(topic);
            }
        }
    }

    @Override
    public List<SubscribeMessage> search(String topic) {
        List<SubscribeMessage> subscribeMessages = new ArrayList<>();
        MAP.forEach((topicFilter, messageMap) -> {
            if (TopicUtil.match(topicFilter, topic)) {
                subscribeMessages.addAll(messageMap.values());
            }
        });

        SHARE_MAP.forEach((topicFilter, message) -> {
            if (TopicUtil.shareMatch(topicFilter, topic)) {
                subscribeMessages.add(message.get());
            }
        });

        return subscribeMessages;
    }

    @Override
    public Integer getTopics() {
        return MAP.size() + SHARE_MAP.size();
    }

    @Override
    public Integer getSubscribes() {
        return MAP.values().stream().mapToInt(Map::size).sum() + SHARE_MAP.values().stream().mapToInt(ShareSubscribeMessage::size).sum();
    }

    /**
     * 分页查询订阅消息列表
     *
     * @param request 参数
     * @return 订阅消息列表
     */
    @Override
    public PageResult<SubscribeMessageResponse> pageSubscribe(PageSubscribeMessageRequest request) {
        List<SubscribeMessage> subscribeMessages = Stream.concat(MAP.values().stream().map(Map::values), SHARE_MAP.values().stream().map(ShareSubscribeMessage::list))
                .flatMap(Collection::stream)
                .filter(subscribe -> !StringUtils.hasText(request.getClientId()) || subscribe.getClientId().contains(request.getClientId()))
                .toList();

        List<SubscribeMessageResponse> responses = subscribeMessages.stream()
                .skip((long) request.getPageSize() * (request.getPageIndex() - 1))
                .limit(request.getPageSize())
                .map(subscribeMessage -> {
                    SubscribeMessageResponse response = new SubscribeMessageResponse();
                    BeanUtils.copyProperties(subscribeMessage, response);

                    return response;
                }).toList();

        return PageResult.success(responses, request.getPageIndex(), request.getPageSize(), subscribeMessages.size());
    }
}
